package com.myflink.day05;

import com.myflink.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @author Shelly An
 * @create 2020/9/22 21:09
 * 有界流的一个小问题
 */
public class Watermark_FileIssue {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .readTextFile("input/sensor-data.log")
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] datas = value.split(",");
                        return new WaterSensor(datas[0], Long.valueOf(datas[1]), Integer.valueOf(datas[2]));
                    }
                })
                .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<WaterSensor>() {
                    @Override
                    public long extractAscendingTimestamp(WaterSensor element) {
                        return element.getTs() * 1000L;
                    }
                });

        sensorDS.keyBy(WaterSensor::getId)
                .timeWindow(Time.seconds(5))
                .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {

                    @Override
                    public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                        out.collect("当前key" +s +", wm= " + context.currentWatermark() + "，一共有"
                                +elements.spliterator().estimateSize()+"条数据");
                    }
                })
                .print();

        /*
        当前keysensor_1, wm= 9223372036854775807，一共有3条数据
        当前keysensor_1, wm= 9223372036854775807，一共有5条数据
        当前keysensor_1, wm= 9223372036854775807，一共有1条数据
        wm 是 long的最大值，为什么呢？不是指定了升序吗？不是应该时间戳减1ms吗？
        因为开窗了，窗口关闭才会输出一条结果。每个窗口，wm到最后都等于long的最大值。为什么？
        窗口1：[20,25)
        窗口2：[25,30)
        窗口3：[30,35)
        文件是有界的，
         */

        env.execute();

    }
}
